package day06

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}
import scala.util.parsing.json.JSON

/**
 * 第3章	数据读取与保存
 *
 * Spark的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。
 *
 * 文件格式分为：Text文件、Json文件、Csv文件、Sequence文件以及Object文件；
 *
 * 文件系统分为：本地文件系统、HDFS以及数据库。
 *
 */
object Data_read_save {

}

/**
 * 3.1	文件类数据读取与保存
 *
 * 3.1.1	Text文件
 * 1）数据读取：textFile(String)
 * 2）数据保存：saveAsTextFile(String)
 * 3）注意：如果是集群路径：hdfs://hadoop102:8020/input/1.txt
 */
object Operate_text {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.读取输入文件
    val inputRdd: RDD[String] = sc.textFile("input/1.txt")

    //4.保存数据
    inputRdd.saveAsTextFile("output")

    // 关闭连接
    sc.stop()
  }
}

/**
 * 3.1.2	Json文件
 * 如果JSON文件中每一行就是一个JSON记录，那么可以通过将JSON文件当做文本文件来读取，然后利用相关的
 * JSON库对每一条数据进行JSON解析。
 *
 * 注意：使用RDD读取JSON文件处理很复杂，同时SparkSQL集成了很好的处理JSON文件的方式，所以应用中多是采用
 * SparkSQL处理JSON文件
 */
object Operate_json {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //导入解析Json所需的包并解析Json
    val rdd: RDD[String] = sc.textFile("input/user.json")
    val resRdd: RDD[Option[Any]] = rdd.map(JSON.parseFull)
    resRdd.collect().foreach(println)

    //修改输入文件格式，执行程序，发现解析失败。原因是一行一行的读取文件
    val rdd1: RDD[String] = sc.textFile("input/user2.json")
    val result: RDD[Option[Any]] = rdd1.map(JSON.parseFull)
    result.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 3.1.3	Sequence文件
 *
 * SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。
 * 在SparkContext中，可以调用sequenceFile[keyClass, valueClass](path)。
 *
 * 注意：SequenceFile文件只针对PairRDD
 */
object Operate_sequence {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val dataRdd: RDD[(Int, Int)] = sc.makeRDD(Array((1, 2), (3, 4), (5, 6)))

    // 保存数据为SequenceFile
    dataRdd.saveAsSequenceFile("output")

    // 读取SequenceFile文件
    val seqRdd: RDD[(Int, Int)] = sc.sequenceFile[Int, Int]("output")
    seqRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 3.1.4	Object对象文件
 *
 * 对象文件是将对象序列化后保存的文件，采用Java的序列化机制。可以通过objectFile[k,v](path)函数接收
 * 一个路径，读取对象文件，返回对应的RDD，也可以通过调用saveAsObjectFile()实现对对象文件的输出。
 * 因为是序列化所以要指定类型。
 */
object Operate_object {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //创建RDD
    val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4))

    //保存数据
    rdd.saveAsObjectFile("output/object")

    //读取数据
    val readRdd: RDD[Int] = sc.objectFile[(Int)]("output/object")

    readRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 3.2	文件系统类数据读取与保存
 *
 * 3.2.1	HDFS
 * Spark的整个生态系统与Hadoop是完全兼容的，所以对于Hadoop所支持的文件类型或者数据库类型，Spark也同样
 * 支持。另外，由于Hadoop的API有新旧两个版本，所以Spark为了能够兼容Hadoop所有的版本，也提供了两套创建
 * 操作接口。对于外部存储创建操作而言，hadoopRDD和newHadoopRDD是最为抽象的两个函数接口
 *
 * 3.2.2	MySQL
 * 支持通过Java JDBC访问关系型数据库。需要通过JdbcRDD进行，示例如下：
 */
object Operate_mysql {
  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.定义连接参数
    val driverClass: String = "com.mysql.jdbc.Driver"
    val url: String = "jdbc:mysql://hadoop102:3306/test"
    val username: String = "root"
    val password: String = "139559"

    //查询数据
    val rdd: JdbcRDD[(Int, String)] = new JdbcRDD(sc, () => {
      Class.forName(driverClass)
      DriverManager.getConnection(url, username, password)
    },
      "select * from student where id >= ? and id <= ?",
      1,
      2,
      1,
      result => (result.getInt(1), result.getString(2))
    )

    println(rdd.count())
    rdd.foreach(println)

    //写数据
    val rdd2: RDD[String] = sc.makeRDD(List("guanyu", "zhangfei", "liubei"))

    //方式一：在循环中创建对象，效率低
    rdd2.foreach {
      name => {
        // 注册驱动
        Class.forName(driverClass)
        // 获取连接
        val conn: Connection = DriverManager.getConnection(url, username, password)
        val sql: String = "insert into student(name) values(?)"
        // 获取数据库操作对象
        val ps: PreparedStatement = conn.prepareStatement(sql)
        // 给参数赋值
        ps.setString(1, name)
        // 执行sql语句
        ps.executeUpdate()
        // 释放资源
        ps.close()
        conn.close()
      }
    }

    // 方式二：在循环外创建。（会报错：java.io.NotSerializableException: com.mysql.jdbc.JDBC4PreparedStatement）
    Class.forName(driverClass)
    val conn: Connection = DriverManager.getConnection(url, username, password)
    val sql: String = "insert into student(name) values(?)"
    val ps: PreparedStatement = conn.prepareStatement(sql)

    rdd2.foreach {
      name => {
        ps.setString(1, name)
        ps.executeUpdate()
      }
    }

    ps.close()
    conn.close()

    //方式三：分区内创建连接对象
    rdd2.foreachPartition {
      datas => {
        Class.forName(driverClass)
        val connection: Connection = DriverManager.getConnection(url, username, password)
        val sql: String = "insert into student(name) values (?)"
        val ps: PreparedStatement = connection.prepareStatement(sql)

        datas.foreach {
          name => {
            ps.setString(1, name)
            ps.executeUpdate()
          }
        }

        ps.close()
        connection.close()
      }
    }

    // 关闭连接
    sc.stop()

  }
}